package com.cmb.lk50.config;

import com.cmb.lk50.common.extractor.QueryParamExtractor;
import com.cmb.lk50.common.receiver.ResponseReceiver;
import com.cmb.lk50.sink.kafka.KafkaKeyGenerator;
import com.cmb.lk50.utils.ReflectUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Author       : 蒋俊钊 80256973
 * Date         : 2020/8/19
 * Copyright    (C) ChinaMerchantsBank
 */
@Configuration
@ConditionalOnProperty(prefix = "access-log", name = "enable", havingValue = "true")
@ConditionalOnWebApplication
public class ComponentConfiguration implements ApplicationContextAware {

    private static final Logger LOGGER = LoggerFactory.getLogger(ComponentConfiguration.class);

    private ApplicationContext applicationContext;

    @Autowired(required = false)
    private List<QueryParamExtractor> extractorBeans;

    @Autowired(required = false)
    private List<ResponseReceiver> receiverBeans;

    @Autowired(required = false)
    private List<KafkaKeyGenerator> keyerBeans;

    @Bean
    public Map<Class, QueryParamExtractor> cacheExtractor() throws ClassNotFoundException, IllegalAccessException, InstantiationException, IOException {
        Map<Class, QueryParamExtractor> cache = cache(QueryParamExtractor.class);
        return merge(cache, extractorBeans);
    }

    @Bean
    public Map<Class, ResponseReceiver> cacheReceiver() throws ClassNotFoundException, IllegalAccessException, InstantiationException, IOException {
        Map<Class, ResponseReceiver> cache = cache(ResponseReceiver.class);
        return merge(cache, receiverBeans);
    }

    @Bean
    public Map<Class, KafkaKeyGenerator> cacheKeyer() throws ClassNotFoundException, IOException, InstantiationException, IllegalAccessException {
        Map<Class, KafkaKeyGenerator> cache = cache(KafkaKeyGenerator.class);
        return merge(cache, keyerBeans);
    }

    // 缓存通过反射创建的工具
    private <T> Map<Class, T> cache(Class<T> target) throws ClassNotFoundException, IllegalAccessException, InstantiationException, IOException {
        Map<String, Object> app = applicationContext.getBeansWithAnnotation(SpringBootApplication.class);
        ConcurrentHashMap<Class, T> cache = new ConcurrentHashMap<>();

        Set<Class<T>> defaultExtractorSub = ReflectUtil.getAllAssignedClass(target);
        for (Class<T> clazz : defaultExtractorSub) {
            cache.put(clazz, clazz.newInstance());
            LOGGER.info("[LogIn] - 实例化[{}]", clazz.getSimpleName());
        }

        for (Object value : app.values()) {
            Set<Class<T>> assignedClass = ReflectUtil.getAllAssignedClass(value.getClass(), target);
            for (Class<T> clazz : assignedClass) {
                cache.put(clazz, clazz.newInstance());
                LOGGER.info("[LogIn] - 实例化[{}]", clazz.getSimpleName());
            }
        }
        return cache;
    }


    // 将spring管理的bean和反射创建的cache合并起来
    private <T> Map<Class, T> merge(Map<Class, T> cache, List<T> bean) {
        if (bean != null) {
            Iterator<T> iterator = bean.iterator();
            while (iterator.hasNext()) {
                T next = iterator.next();
                if (cache.containsKey(next.getClass())) {
                    cache.put(next.getClass(), next);
                    iterator.remove();
                }
            }
        }
        return cache;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
